package com.markus.code.Thread;

import com.markus.code.Thread.pool.MyThreadPool;

import java.util.Random;
import java.util.concurrent.CompletableFuture;
import java.util.concurrent.ExecutionException;
import java.util.concurrent.ThreadPoolExecutor;
import java.util.concurrent.TimeUnit;
import java.util.function.BiConsumer;
import java.util.function.BiFunction;
import java.util.function.Consumer;
import java.util.function.Function;

/**
 * Author:markusZhang
 * Date:Create in 2020/8/17 11:33
 * todo: CompletableFuture异步任务测试
 */
public class CompletableFutureDemo {

    /**
     * 无返回值
     */
    public static void runAsync() throws ExecutionException, InterruptedException {
        MyThreadPool pool = new MyThreadPool();
        ThreadPoolExecutor poolExecutor = pool.threadPoolExecutor();
        CompletableFuture<Void> future = CompletableFuture.runAsync(()->{
            try {
                TimeUnit.SECONDS.sleep(3);
            } catch (InterruptedException e) {
                e.printStackTrace();
            }
        },poolExecutor);
        System.out.println(future.get());
    }

    /**
     * 有返回值
     */
    public static void supplyAsync() throws ExecutionException, InterruptedException {
        MyThreadPool pool = new MyThreadPool();
        ThreadPoolExecutor poolExecutor = pool.threadPoolExecutor();
        CompletableFuture<Integer> future = CompletableFuture.supplyAsync(()->{
            try {
                TimeUnit.SECONDS.sleep(3);
            } catch (InterruptedException e) {
                e.printStackTrace();
            }
            return new Integer(1);
        },poolExecutor);
        System.out.println(future.get());
    }

    /**
     * 计算结果完时的回调方法
     * 当CompletableFuture完成后，或者抛出异常，执行特定的方法
     * public CompletableFuture<T> whenComplete(BiConsumer<? super T, ? super Throwable> action);
     * public CompletableFuture<T> whenCompleteAsync(BiConsumer<? super T, ? super Throwable> action);
     * public CompletableFuture<T> whenCompleteAsync(BiConsumer<? super T, ? super Throwable> action, Executor executor);
     * public CompletableFuture<T> exceptionally(Function<Throwable, ? extends T> fn);
     */
    public static void whenComplete(){
        MyThreadPool pool = new MyThreadPool();
        ThreadPoolExecutor poolExecutor = pool.threadPoolExecutor();
        CompletableFuture<Void> future = CompletableFuture.runAsync(()->{
            try {
                TimeUnit.SECONDS.sleep(3);
            } catch (InterruptedException e) {
                e.printStackTrace();
            }
            if (new Random().nextInt()%2 == 0){
                int i = 12/0;
            }
            System.out.println("run end ......");
        },poolExecutor);
        future.whenCompleteAsync(new BiConsumer<Void, Throwable>() {
            @Override
            public void accept(Void aVoid, Throwable throwable) {
                System.out.println("执行完成");
            }
        });
        future.exceptionally(new Function<Throwable, Void>() {
            @Override
            public Void apply(Throwable throwable) {
                System.out.println("执行失败");
                return null;
            }
        });
    }

    /**
     * thenApply 方法
     * 当一个线程依赖另一个线程时，可以使用 thenApply 方法来把这两个线程串行化。
     * Function<? super T,? extends U>
     * T：上一个任务返回结果的类型
     * U：当前任务的返回值类型
     */
    public static void thenApply() throws ExecutionException, InterruptedException {
        MyThreadPool pool = new MyThreadPool();
        ThreadPoolExecutor poolExecutor = pool.threadPoolExecutor();
        CompletableFuture<Integer> future = CompletableFuture.supplyAsync(()->{
            try {
                TimeUnit.SECONDS.sleep(2);
            } catch (InterruptedException e) {
                e.printStackTrace();
            }
            return new Integer(2);
        },poolExecutor);
        CompletableFuture<Integer> thenApply = future.thenApply(new Function<Integer, Integer>() {
            @Override
            public Integer apply(Integer integer) {
                return integer + 1;
            }
        });
        System.out.println("原始值");
        System.out.println(future.get());
        System.out.println("后来值");
        System.out.println(thenApply.get());
    }

    /**
     * handle 是执行任务完成时对结果的处理。
     * handle 方法和 thenApply 方法处理方式基本一样。不同的是 handle 是在任务完成后再执行，还可以处理异常的任务。
     * thenApply 只可以执行正常的任务，任务出现异常则不执行 thenApply 方法。
     */
    public static void handle() throws ExecutionException, InterruptedException {
        MyThreadPool pool = new MyThreadPool();
        ThreadPoolExecutor poolExecutor = pool.threadPoolExecutor();
        CompletableFuture<Integer> future = CompletableFuture.supplyAsync(()->{
            try {
                TimeUnit.SECONDS.sleep(2);
            } catch (InterruptedException e) {
                e.printStackTrace();
            }
            int i = 1/0;
            return new Integer(2);
        },poolExecutor);
        CompletableFuture<Integer> handle = future.handle(new BiFunction<Integer, Throwable, Integer>() {
            @Override
            public Integer apply(Integer integer, Throwable throwable) {
                if (throwable != null) {
                    System.out.println(throwable.getMessage());
                } else {
                    return integer + 1;
                }
                return null;
            }
        });
        System.out.println(handle.get());
    }

    /**
     *  thenAccept 消费处理结果
     * 接收任务的处理结果，并消费处理，无返回结果。
     */
    public static void thenAccept(){
        MyThreadPool pool = new MyThreadPool();
        ThreadPoolExecutor poolExecutor = pool.threadPoolExecutor();
        CompletableFuture<Integer> future = CompletableFuture.supplyAsync(()->{
            try {
                TimeUnit.SECONDS.sleep(2);
            } catch (InterruptedException e) {
                e.printStackTrace();
            }
            return new Integer(2);
        },poolExecutor);
        CompletableFuture<Void> thenAccept = future.thenAccept(new Consumer<Integer>() {
            @Override
            public void accept(Integer integer) {
                System.out.println("我的任务就是打印上面任务的结果：" + integer);
            }
        });
    }

    /**
     * thenRun 方法
     * 跟 thenAccept 方法不一样的是，
     * 不关心任务的处理结果。只要上面的任务执行完成，就开始执行 thenRun 。
     */
    public static void thenRun(){
        MyThreadPool pool = new MyThreadPool();
        ThreadPoolExecutor poolExecutor = pool.threadPoolExecutor();
        CompletableFuture<Void> future = CompletableFuture.supplyAsync(()->{
            try {
                TimeUnit.SECONDS.sleep(2);
            } catch (InterruptedException e) {
                e.printStackTrace();
            }
            return new Integer(2);
        },poolExecutor).thenRun(()-> {
            System.out.println("thenRun...");
        });
    }
    public static void main(String[] args) throws ExecutionException, InterruptedException {
//        System.out.println("runAsync:");
//        CompletableFutureDemo.runAsync();
//        System.out.println("supplyAsync");
//        CompletableFutureDemo.supplyAsync();
//        System.out.println("计算完成的回调");
//        CompletableFutureDemo.whenComplete();
//        System.out.println("使任务串行化");
//        CompletableFutureDemo.thenApply();
//        System.out.println("handle方法");
//        CompletableFutureDemo.handle();
//        System.out.println("thenAccept方法");
//        CompletableFutureDemo.thenAccept();
        System.out.println("thenRun方法");
        CompletableFutureDemo.thenRun();
    }
}
